package 实训二;

import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyClickHouseUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArIsNewPageWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> pageDs = env.addSource(MyKafkaUtil.
                getFlinkKafkaConsumer("dwd_traffic_page_log", "xxxx"));
        DataStreamSource<String> uvDs = env.addSource(MyKafkaUtil.
                getFlinkKafkaConsumer("dwd_traffic_unique_visitor_detail", "xxxx"));
        DataStreamSource<String> ujeDs = env.addSource(MyKafkaUtil.
                getFlinkKafkaConsumer("dwd_traffic_user_jump_detail", "xxxx"));


        // 你确定不用运行jar
        SingleOutputStreamOperator<TrafficPageViewBean> uvmap = uvDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null,
                        vc, ch, ar, isNew, 1L,//uvCt=1L：表示这是一个独立访客（计数 +1）
                        0L, 0L, 0L, 0L, jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> ujemap = ujeDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null,
                        vc, ch, ar, isNew, 0L,
                        0L, 0L, 0L, 1L, jsonObject.getLong("ts"));//ujCt=1L 表示一次跳转事件。
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> pageMap = pageDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                Long sv=0L;
                // TODO: 2025/9/17 // 如果是首次访问，则算作“会话开始”
                if (last_page_id == null){
                    sv=1L;
                }
            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L,
                    sv, 1L, page.getLong("during_time"), 0L, jsonObject.getLong("ts"));//pvCt=1：每次页面加载都计为一次 PV
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> union = pageMap.union(uvmap, ujemap)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }));
        union.print("数据===========>");

        SingleOutputStreamOperator<TrafficPageViewBean> reduce = union.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                        return Tuple4.of(trafficPageViewBean.getVc(), trafficPageViewBean.getCh(), trafficPageViewBean.getAr(), trafficPageViewBean.getIsNew());
                    }
                })
                // TODO: 2025/9/17 使用滚动事件时间窗口，每 10 秒一个窗口。允许迟到数据最多 10 秒。
                .window(TumblingEventTimeWindows.of(Time.seconds(10L))).allowedLateness(Time.seconds(10L))
                // TODO: 2025/9/17 对每个窗口内同一 key 的数据做累加： pvCt, uvCt, svCt, ujCt, durSum 都累加。
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean t1, TrafficPageViewBean t2) throws Exception {
                        t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                        t1.setUvCt(t1.getUvCt() + t2.getUvCt());
                        t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                        t1.setUjCt(t1.getUjCt() + t2.getUjCt());
                        t1.setDurSum(t1.getDurSum() + t2.getDurSum());
                        return t1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                        // TODO: 2025/9/17 取出窗口中任意一条记录（input.iterator().next()）作为结果
                        TrafficPageViewBean next = input.iterator().next();
                        Timestamp start = new Timestamp(window.getStart());
                        Timestamp end = new Timestamp(window.getEnd());
                        next.setStt(start);
                        next.setEdt(end);
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        reduce.print();
        reduce.addSink(MyClickHouseUtil.getSinkFunction("insert into gmall.dws_traffic_vc_ch_page_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
